/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package maxent.mr.io;

import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import stanford.tagger.TagToken;

/**
 *
 * @author hadoop
 */
public class TagTokenReader implements HDFSReader{

    private String inputDir;   
    private TagToken tagToken;

    public TagTokenReader() {
        tagToken = new TagToken();
    }

    public TagTokenReader(String inputDir) {
        this.inputDir = inputDir;        
        tagToken = new TagToken();
    }

    public String getInputDir() {
        return inputDir;
    }

    public void setInputDir(String inputDir) {
        this.inputDir = inputDir;
    }

    public TagToken getTagToken() {
        return tagToken;
    }

    public void read(){
        try {
            Configuration conf = new Configuration();
            FileSystem fs = FileSystem.get(conf);
            FileStatus[] inputFiles = fs.listStatus(new Path(inputDir));

            for (int i = 0; i < inputFiles.length; i++) {
                // don't process the log files
                if (!inputFiles[i].getPath().getName().equalsIgnoreCase("_logs")) {
                    // input stream
                    FSDataInputStream in = fs.open(inputFiles[i].getPath());

                   String lines;
                   String tag;
                   String words[];

                    while((lines=in.readLine()) != null) {
                        try{

                            tag = lines.split("\t")[0];                            
                            words = getWords(lines.split("\t")[1]);
                            tagToken.add(tag, words);
                            
                        }catch(Exception e){
                            e.getMessage();
                        }
                    }
                    in.close();
                }
                
            }
        } catch (IOException iOException) {
            iOException.printStackTrace();
        }
    }

    private String[] getWords(String line){
        
        return line.split(" ");
    }

    public void print(){

        Iterator<String> tagIt = tagToken.get().keySet().iterator();

        while(tagIt.hasNext()){
            String tag = tagIt.next();
            System.out.print(tag+": ");
            Iterator<String> wordIt = tagToken.get().get(tag).iterator();

            while(wordIt.hasNext()){
                System.out.print(wordIt.next()+" ");
            }
            System.out.println();
        }
        
    }
}
